package geerpc

import (
	"encoding/json"
	"errors"
	"fmt"
	"gitee.com/bankarian/gee-rpc/codec"
	"log"
	"net"
	"sync"
)

type Call struct {
	Seq           uint64
	ServiceMethod string      // format "<service>.<method>"
	Args          interface{} // arguments to the function
	Reply         interface{} // reply from the function
	Error         error
	Done          chan *Call // support asynchronous
}

func (call *Call) done() {
	call.Done <- call
}

type Client struct {
	cc       codec.Codec
	opt      *codec.Option
	sending  sync.Mutex
	header   codec.Header
	mutex    sync.Mutex
	seq      uint64
	pending  map[uint64]*Call
	closing  bool // user stop
	shutdown bool // server stop
}

var ErrShutdown = errors.New("connection is shut down")

func (client *Client) Close() error {
	client.mutex.Lock()
	defer client.mutex.Unlock()
	if client.closing {
		return ErrShutdown
	}
	client.closing = true
	return client.cc.Close()
}

// check if the client does work
func (client *Client) IsAvailable() bool {
	client.mutex.Lock()
	defer client.mutex.Unlock()
	return !client.shutdown && !client.closing
}

func (client *Client) isAvailable() bool {
	return !client.shutdown && !client.closing
}

func (client *Client) registerCall(call *Call) (uint64, error) {
	client.mutex.Lock()
	defer client.mutex.Unlock()
	if !client.isAvailable() {
		return 0, ErrShutdown
	}
	call.Seq = client.seq
	client.pending[call.Seq] = call
	client.seq++
	return call.Seq, nil
}

// remove and return the pending call
func (client *Client) removeCall(seq uint64) *Call {
	client.mutex.Lock()
	defer client.mutex.Unlock()
	call, _ := client.pending[seq]
	delete(client.pending, seq)
	return call
}

// terminate pending calls when server encounters an error
func (client *Client) terminateCalls(err error) {
	client.sending.Lock()
	client.mutex.Lock()
	defer client.sending.Unlock()
	defer client.mutex.Unlock()
	client.shutdown = true
	for _, call := range client.pending {
		call.Error = err
		call.done()
	}
}

/*
three cases when receiver from the server
1. call doesn't exist, but server still handled the request

2. call exists, but server encountered error when handling it

3. call exists, all as well, just read the body from reply
*/
func (client *Client) receive() {
	var err error
	for err == nil {
		var header codec.Header
		if err = client.cc.ReadHeader(&header); err != nil {
			break
		}
		call := client.removeCall(header.Seq)
		switch {
		case call == nil:
			err = client.cc.ReadBody(nil)
		case header.Error != "":
			call.Error = fmt.Errorf(header.Error)
			err = client.cc.ReadBody(nil)
			call.done()
		default:
			err = client.cc.ReadBody(call.Reply)
			if err != nil {
				call.Error = errors.New("reading body " + err.Error())
			}
			call.done()
		}
	}
	// error occurs, terminates all pending calls
	client.terminateCalls(err)
}

// supports concurrency
func (client *Client) send(call *Call) {
	client.sending.Lock() // guarantee a complete request
	defer client.sending.Unlock()

	// register call
	seq, err := client.registerCall(call)
	if err != nil {
		call.Error = err
		call.done()
		return
	}

	// prepare request header
	client.header.ServiceMethod = call.ServiceMethod
	client.header.Seq = seq
	client.header.Error = ""

	// encode and send the request
	if err := client.cc.Write(&client.header, call.Args); err != nil {
		call := client.removeCall(seq)
		// call may be nil, usually means that Write partially failed
		if call != nil {
			call.Error = err
			call.done()
		}
	}
}

// Go invokes the function asynchronously,
// returning the Call.
//
// Go supports asynchronous & concurrency
func (client *Client) Go(serviceMethod string, args, reply interface{}, done chan *Call) *Call  {
	if done == nil {
		done = make(chan *Call, 10)
	} else if cap(done) == 0 {
		log.Panic("rpc client: done channel is unbuffered")
	}
	call := &Call{
		ServiceMethod: serviceMethod,
		Args: args,
		Reply: reply,
		Done: done,
	}
	client.send(call)
	return call
}

// Call wraps up the *Client.Go,
// invokes the named function, waits for it to complete,
// returns on error
func (client *Client) Call(serviceMethod string, args, reply interface{}) error {
	call := <- client.Go(serviceMethod, args, reply, make(chan *Call, 1)).Done
	return call.Error
}

func NewClient(conn net.Conn, opt *codec.Option) (*Client, error) {
	fn, ok := codec.NewCodecFuncMap[opt.CodecType]
	if !ok {
		err := fmt.Errorf("invalid codec type %s", opt.CodecType)
		log.Println("rpc client: codec error: ", err)
		return nil, err
	}
	// send options to server
	// so as to reach an agreement on encoding
	if err := json.NewEncoder(conn).Encode(opt); err != nil {
		log.Println("rpc client: options error: ", err)
		_ = conn.Close()
		return nil, err
	}
	return newClientCodec(fn(conn), opt), nil
}

func newClientCodec(cc codec.Codec, opt *codec.Option) *Client {
	client := &Client{
		seq:     1, // seq starts from 1, 0 means invalid call
		cc:      cc,
		opt:     opt,
		pending: make(map[uint64]*Call),
	}
	go client.receive()
	return client
}

func parseOptions(opts ...*codec.Option) (*codec.Option, error) {
	// didn't get an option
	if len(opts) == 0 || opts[0] == nil {
		return codec.DefaultOption, nil
	}
	if len(opts) != 1 {
		return nil, errors.New("to many options, must [0, 1]")
	}
	opt := opts[0]
	opt.MagicNumber = codec.DefaultOption.MagicNumber
	if opt.CodecType == "" {
		opt.CodecType = codec.DefaultOption.CodecType
	}
	return opt, nil
}

func Dial(network, address string, opts ...*codec.Option) (client *Client, err error) {
	opt, err := parseOptions(opts...)
	if err != nil {
		return nil, err
	}
	conn, err := net.Dial(network, address)
	if err != nil {
		return nil, err
	}
	// close the connection if the client is nil
	defer func() {
		if client == nil {
			_ = conn.Close()
		}
	}()
	return NewClient(conn, opt)
}
